/**
 * Licensed under:
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * 
 * @author Emanuele Gherardini (e.gherardini@yahoo.it / e.gherardini@hotmail.com)
 */
package org.apache.camel.component.coherence;

import static org.apache.camel.component.coherence.CoherenceConstants.QUEUE_PREFIX;
import static org.apache.camel.component.coherence.CoherenceConstants.TOPIC_PREFIX;
import static org.apache.camel.util.ObjectHelper.after;

import java.util.Map;

import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.component.coherence.queue.CoherenceQueueEndpoint;
import org.apache.camel.component.coherence.topic.CoherenceTopicEndpoint;
import org.apache.camel.impl.DefaultComponent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 
 * TODO: add concurrentConsumers option
 * 
 * 
 * @author gheema
 */
public class CoherenceComponent extends DefaultComponent {

	private static final Logger log = LoggerFactory.getLogger(CoherenceComponent.class);
	private CoherenceClusterConfiguration clusterConfiguration;
	
	public CoherenceComponent() {
		log.info("Creating CoherenceComponent using configuration: DEFAULT");
    }
	
	/**
	 * Builds the component using the given 
	 * cluster configuration.
	 * @param clusterName
	 */
	public CoherenceComponent(CoherenceClusterConfiguration clusterConfiguration) {
		this.clusterConfiguration = clusterConfiguration;
		log.info("Creating CoherenceComponent using configuration: " + clusterConfiguration);
		clusterConfiguration.applyClusterConfiguration();
	}

	/**
	 * Constructs an endpoint using default values for
	 * the cluster configuration. 
	 * 
	 * It is strongly suggested to provide your own configuration.
	 * 
	 * @param context
	 */
    public CoherenceComponent(CamelContext context) {
        super(context);
        log.info("Creating CoherenceComponent using configuration: DEFAULT");
    }
    
    public CoherenceComponent(CamelContext context, CoherenceClusterConfiguration clusterConfiguration) {
    	super(context);
    	this.clusterConfiguration = clusterConfiguration;
    	log.info("Creating CoherenceComponent using configuration: " + clusterConfiguration);
    	clusterConfiguration.applyClusterConfiguration();
    }
    
    /**
     * Set the system property to let coherence use the 
     * override file defined
     */
    {
		System.setProperty("tangosol.coherence.override", "messaging-tangosol-coherence-override.xml");
	}
	
	@Override
	protected Endpoint createEndpoint(String uri, String remaining,
			Map<String, Object> parameters) throws Exception {
		
		Endpoint endpoint = null;
		
		boolean durableSubscriber = getAndRemoveParameter(parameters, "durableSubscribers", Boolean.class, false);
		int concurrentConsumers = getAndRemoveParameter(parameters, "concurrentConsumers", Integer.class, 1);
		
		if (remaining.startsWith(QUEUE_PREFIX.getValue())) {
			remaining = after(remaining, QUEUE_PREFIX.getValue());
			endpoint = this.createQueueEndpoint(uri, remaining, durableSubscriber,concurrentConsumers);
		}
		
		if(remaining.startsWith(TOPIC_PREFIX.getValue())) {
			remaining = after(remaining, TOPIC_PREFIX.getValue());
			endpoint = this.createTopicEndpoint(uri, remaining, durableSubscriber, concurrentConsumers);
		}
		
		return endpoint;
		
	}
	
	private CoherenceQueueEndpoint createQueueEndpoint(String uri, String queueName, boolean durableSubscriber, int concurrentConsumers) {
		return new CoherenceQueueEndpoint(uri, this, queueName, durableSubscriber, concurrentConsumers);
	}
	
	private CoherenceTopicEndpoint createTopicEndpoint(String uri, String topicName, boolean durableSubscriber, int concurrentConsumers) {
		return new CoherenceTopicEndpoint(uri, this, topicName, durableSubscriber, concurrentConsumers);
	}
	
	public CoherenceClusterConfiguration getClusterConfiguration() {
		return clusterConfiguration;
	}

	public void setClusterConfiguration(
			CoherenceClusterConfiguration clusterConfiguration) {
		this.clusterConfiguration = clusterConfiguration;
	}
	

	@Override
	public String toString() {
		return "CoherenceComponent [clusterConfiguration="
				+ clusterConfiguration + "]";
	}
	
	

}
